package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo04DianXin {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val dxDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("mdn String, grid_id String, city_id String, county_id String, duration Int, start_time String, end_time String, pt String")
      .load("spark/data/dianxin.csv")

    //    dxDF.show()
    // 1、统计每座城市停留时间超过两小时的游客人数
    dxDF
      // 统计每个游客在每个城市的停留时间之和
      .groupBy($"mdn", $"city_id")
      .agg(sum($"duration") as "sum_duration")
      .where($"sum_duration" > 2 * 60)
      // 统计人数
      .groupBy($"city_id")
      .agg(count($"mdn") as "cnt")
    //      .show()

    // 2、统计每座城市停留时间最久的前三个区县
    dxDF
      // 每座城市每个区县的游客停留时间之和
      .groupBy($"city_id", $"county_id")
      .agg(sum($"duration") as "sum_duration")
      .withColumn("rn", row_number() over Window.partitionBy($"city_id").orderBy($"sum_duration".desc))
      .where($"rn" <= 3)
    //      .show()

    // 3、统计每座城市游客人数最多的前三个区县
    dxDF
      // 每座城市每个区县的游客人数
      .groupBy($"city_id", $"county_id")
      .agg(countDistinct($"mdn") as "cnt")
      .withColumn("rn", row_number() over Window.partitionBy($"city_id").orderBy($"cnt".desc))
      .where($"rn" <= 3)
    //      .show()

    // 4、统计每个用户在同一个区县内相邻两条位置记录之间的时间间隔
    dxDF
      .select($"mdn", $"county_id", $"start_time", $"end_time")
      .withColumn("last_end_time", lag($"end_time", 1) over Window.partitionBy($"mdn", $"county_id").orderBy($"start_time"))
      .withColumn("interval", round(coalesce(
        (unix_timestamp($"start_time", "yyyyMMddHHmmss")
          - unix_timestamp($"last_end_time", "yyyyMMddHHmmss")
          ) / 60, expr("0")
      ), 2)
      )
      // 整理数据
      .select($"mdn", $"county_id", $"start_time", $"end_time", $"interval")
    //      .show()

    // 5、统计每个网格内停留时间降序排名，最长、最短以及平均停留时间
    dxDF
      .select($"grid_id", $"duration")
      .withColumn("rn", row_number() over Window.partitionBy($"grid_id").orderBy($"duration".desc))
      .withColumn("max_duration", max($"duration") over Window.partitionBy($"grid_id"))
      .withColumn("min_duration", min($"duration") over Window.partitionBy($"grid_id"))
      .withColumn("avg_duration", round(avg($"duration") over Window.partitionBy($"grid_id"), 2))
    //      .show()

  }


}
